package com.ads;

import com.alibaba.fastjson.JSONObject;
import com.utils.KafkaUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Tuple;

/**
 * 统计每个地区、品类的销售额、销售量、客单价以及复购率，并将结果写入到 ClickHouse 中
 */
public class nine4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从kafka主题读取order_info和order_detail的宽表使用广播关联sku_info的大表
        DataStreamSource<String> map = env.addSource(KafkaUtils.createConsumer("ads_wide_db", "iKun"));

        SingleOutputStreamOperator<JSONObject> stream = map.map(x -> {
            return JSONObject.parseObject(x);
        });
        stream.print();
        //统计每种商品的销售额、销售量和客单价，并将结果写入到 ClickHouse 中（10分）
//        KeyedStream<JSONObject, Tuple2<String, String>> jsonObjectTuple2KeyedStream = stream.keyBy(x -> {
//            String provinceId = x.getJSONObject("info").getJSONObject("after").getString("province_id");
//            String id = x.getJSONObject("sku_id").getJSONObject("after").getString("id");
//            return Tuple2.of(provinceId, id);
//        });

        KeyedStream<JSONObject, Tuple2<String, String>> jsonObjectTuple2KeyedStream = stream.keyBy(new KeySelector<JSONObject, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(JSONObject jsonObject) throws Exception {
                String provinceId = jsonObject.getJSONObject("info").getJSONObject("after").getString("province_id");
                String id = jsonObject.getJSONObject("sku_id").getJSONObject("after").getString("id");
                return Tuple2.of(provinceId, id);
            }
        });




        SingleOutputStreamOperator<String> processStream = jsonObjectTuple2KeyedStream.process(new KeyedProcessFunction<Tuple2<String,String>, JSONObject, String>() {
            ValueState<Double> moneyState;
            ValueState<Double> numState;
            JSONObject jo;
            Integer a;
            ValueState<Integer> fuGouState;
            ValueState<Integer> allState;
            @Override
            public void open(Configuration parameters) throws Exception {
                //监控状态
                ValueStateDescriptor<Integer> fuGou = new ValueStateDescriptor<>("fugou", Integer.class);
                ValueStateDescriptor<Integer> all = new ValueStateDescriptor<>("all", Integer.class);
                fuGouState = getRuntimeContext().getState(fuGou);
                allState = getRuntimeContext().getState(all);
                ValueStateDescriptor<Double> money = new ValueStateDescriptor<>("price", Double.class);
                ValueStateDescriptor<Double> num = new ValueStateDescriptor<>("num", Double.class);
                moneyState = getRuntimeContext().getState(money);
                numState = getRuntimeContext().getState(money);
                //复购率  购买两次以上的用户/所有购买的人
                 jo = new JSONObject();
                  a=0;
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<Tuple2<String,String>, JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //统计每种商品的销售额、销售量和客单价
                //销售额、销售量、客单价以及复购率
                Double money = moneyState.value();
                if (money == null) {
                    money = 0.0;
                    moneyState.update(money);
                }
                Double num = numState.value();
                if (num == null) {
                    num = 0.0;
                    numState.update(num);
                }
                //获取用户的id,每来一个人,就将其加入jo中
                Integer fuGouValue = fuGouState.value();
                Integer all = allState.value();
                String userId = jsonObject.getJSONObject("info").getJSONObject("after").getString("user_id");
                //先判断jo,如果包含这个key,就使值+1,即复购的人数,如果不包含,就将
                if(fuGouValue==null){
                    fuGouValue=0;
                }
                if(all==null){
                    all=0;
                }
                if(jo.containsKey(userId)){
                    fuGouValue++;
                    fuGouState.update(fuGouValue);
                }
                all++;
                allState.update(all);
                jo.put(userId,"1");
                //销售额=销售单价x销售量
                Double price = jsonObject.getJSONObject("detail").getJSONObject("after").getDouble("order_price");
                Double count = jsonObject.getJSONObject("detail").getJSONObject("after").getDouble("sku_num");
                money=money+(price*count);
                moneyState.update(money);
                //销售量
                num=num+ count;
                numState.update(num);
                //客单价
                Double singlePrice = money / num;
                collector.collect(context.getCurrentKey() + "-" + "销售额:" + money + "," + "销售量:" + num + "," + "客单价:" + singlePrice+",复购率:"+fuGouValue*1.0/all);
            }
        });
        processStream.print();
        env.execute();

    }
}
